package com.tlb.springboot.starter.rocketmq.impl;

import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.MessageExt;
import com.tlb.springboot.starter.rocketmq.annotation.RocketMQListen;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Method;
import java.util.List;

/**
 * @author yuaneg 2018/8/30
 */
@Slf4j
public class DefaultMessageListenerOrderly extends AbstractMessageListener implements MessageListenerOrderly {


    public DefaultMessageListenerOrderly(Object bean, Method method, RocketMQListen rocketMQListen) {
        super(bean, method, rocketMQListen);
    }

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        for (MessageExt messageExt : msgs) {
            try {
                invokeMethod(messageExt);
            } catch (Exception e) {
                context.setSuspendCurrentQueueTimeMillis(rocketMQListen.suspendCurrentQueueTimeMillis());
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
